package com.gitee.ykr.lightweight.rpc.protocol.client;

import com.gitee.ykr.lightweight.rpc.common.entity.NettyRpcResponse;
import com.gitee.ykr.lightweight.rpc.common.entity.NettyRpcRequest;
import com.gitee.ykr.lightweight.rpc.protocol.serializer.RpcDecoder;
import com.gitee.ykr.lightweight.rpc.protocol.serializer.RpcEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author YKR
 * @Describe
 * @date 2021/8/16
 */
@NoArgsConstructor
@Data
public class RpcClient extends SimpleChannelInboundHandler<NettyRpcResponse> {

    private String ip;

    private int port;

    private NettyRpcResponse nettyRpcResponse;

    public RpcClient(String ip,int port){
        this.ip=ip;
        this.port=port;
    }

    /**
     * 向服务端发送请求，返回响应
     * @param nettyRpcRequest
     * @return nettyRpcResponse
     */
    public NettyRpcResponse send(NettyRpcRequest nettyRpcRequest){
        //创建事件循环
        EventLoopGroup group = new NioEventLoopGroup();
        //创建客户端引导
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new RpcEncoder(NettyRpcRequest.class)); // 编码 RPC 请求
                pipeline.addLast(new RpcDecoder(NettyRpcResponse.class)); // 解码 RPC 响应
                pipeline.addLast(RpcClient.this); // 处理 RPC 响应
            }
        });
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        try {
            // 连接 RPC 服务器
            ChannelFuture future = bootstrap.connect(ip, port).sync();
            // 写入 RPC 请求数据并关闭连接
            Channel channel = future.channel();
            channel.writeAndFlush(nettyRpcRequest).sync();
            channel.closeFuture().sync();
            // 返回 RPC 响应对象
            return nettyRpcResponse;
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException("客户端网络异常！");
        }finally {
            //关闭事件循环
            group.shutdownGracefully();
        }

    }

    @Override
    public void channelRead0(ChannelHandlerContext channelHandlerContext, NettyRpcResponse nettyRpcResponse) throws Exception {
        this.nettyRpcResponse = nettyRpcResponse;
    }

    /**
     * 异常处理器，处理netty运行中的异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        ctx.close();
        throw new RuntimeException(String.format("netty出现异常:%s",cause));
    }
}
